package com.atguigu.gmall.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DwdInteractionCommentInfo
 * Description:
 *
 * @Create 2024/4/20 11:19
 */
public class DwdInteractionCommentInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdInteractionCommentInfo().start(10012,4,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //TODO kafka读topic_db数据，，创建动态表
        readOdsDb(tableEnv,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
        //tableEnv.executeSql("select * from topic_db").print();

        //TODO 过滤评论数据
        Table commentInfo = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['appraise'] appraise,\n" +
                "`data`['comment_txt'] comment_txt,\n" +
                "ts,\n" +
                "pt\n" +
                "from topic_db where `table` = 'comment_info' and `type` = 'insert'");

        //commentInfo.execute().print();
        tableEnv.createTemporaryView("comment_info",commentInfo);

        //TODO Hbase中查询字典数据，创建动态表
        readBaseDic(tableEnv);

        //tableEnv.executeSql("select * from base_dic").print();

        //TODO 将评论和字典表数据进行关联
        Table joinedTable = tableEnv.sqlQuery("SELECT id,\n" +
                "       user_id,\n" +
                "       sku_id,\n" +
                "       appraise,\n" +
                "       dic.dic_name appraise_name,\n" +
                "       comment_txt,\n" +
                "       ts\n" +
                "FROM comment_info AS c\n" +
                "         JOIN base_dic FOR SYSTEM_TIME AS OF c.pt AS dic\n" +
                "ON c.appraise = dic.dic_code;");
        //joinedTable.execute().print();

        //TODO 将关联结果写到kafka主题
        tableEnv.executeSql("CREATE TABLE " + Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO + "\n" +
                "(\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    sku_id string,\n" +
                "    appraise string,\n" +
                "    appraise_name string,\n" +
                "    comment_txt string,\n" +
                "    ts bigint,\n" +
                //指定主键，upsert kafka有更新操作，主键保证幂等
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ")\n" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));

        joinedTable.executeInsert(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
    }




}
